package com.task.scheduling.job;

import com.task.scheduling.constants.Constants;
import com.task.scheduling.dao.TaskSchedulingDao;
import com.task.scheduling.enums.DispatcherStatus;
import com.task.scheduling.enums.TaskStatus;
import com.task.scheduling.model.TaskSchedulingModel;
import com.task.scheduling.runner.ProcessorRunner;
import com.task.scheduling.service.DispatcherHolder;
import com.task.scheduling.service.FutureHolder;
import com.task.scheduling.util.HostUtil;
import com.task.scheduling.util.SpringContextUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;


/**
 * @program: task-scheduling
 * @description: 执行新的任务
 * @author: zhouliangliang
 * @create: 2018-11-14 14:25
 **/
public class RunNewJob implements Runnable {

    private Logger log = LoggerFactory.getLogger(RunNewJob.class);

    private int poolSize = 20;
    //固定线程池
    private ExecutorService executor = Executors.newFixedThreadPool(poolSize);

    final Semaphore semp = new Semaphore(poolSize);

    //用来控制停止任务
    private static volatile boolean flag = true;
    private TaskSchedulingDao taskSchedulingDao;
    public RunNewJob(TaskSchedulingDao taskSchedulingDao) {
        this.taskSchedulingDao = taskSchedulingDao;
    }

    /**
     * 1.首先获取任务
     * 2.更新任务为锁定状态
     * 3.检查任务是否为自己锁定
     * 4.检查成功之后执行任务
     */
    @Override
    public void run() {
        try {
            //睡眠随机时间，以防止多台机器争抢任务
            Thread.sleep(new Random().nextInt(10) * 1000);

            //机器所属环境执行不了的任务
            List<Long> ids = new LinkedList<Long>();
            while (flag) {
                if(DispatcherHolder.getDispatcherStatus() == DispatcherStatus.open.getStatus()) {
                    semp.acquire();
                    //首先获取任务
                    TaskSchedulingModel task = taskSchedulingDao.getNewTask(new Date(), 0, ids);
                    if(task != null) {
                        //验证一下task的运行时间是不是符合要求(在当前时间之前)
                        Date execTime = task.getExecutionTime();
                        Date now = new Date();
                        if(now.before(execTime)) {
                            //现在时间在任务开始执行时间之前，说明不满足执行条件
                            log.info("现在时间为：{} , 任务执行时间为: {}, 不满足执行条件，放弃执行任务！", now, execTime);
                            Thread.sleep(Constants.runNewJobInterval);
                            semp.release();
                            continue;
                        }

                        String ip = HostUtil.getIp();
                        //如果获取到任务的话，尝试锁定任务
                        int count = taskSchedulingDao.lockTaskSchedule(task.getId(),  ip);
                        log.info("锁定数量：" + count);
                        if (count == 1) {
                            //锁定任务成功，则等待一段时间，验证是不是锁定成功
                            this.idle();
                            TaskSchedulingModel lockTask = taskSchedulingDao.getTaskById(task.getId());
                            if(lockTask.getTaskStatus() == TaskStatus.LOCKED.getCode() && lockTask.getExecutor().equalsIgnoreCase(ip)) {
                                TaskSchedulingModel upModel = new TaskSchedulingModel();
                                upModel.setId(task.getId());
                                upModel.setTaskStatus(TaskStatus.RUNNING.getCode());
                                int size = taskSchedulingDao.updateTaskSchedule(upModel);
                                if(size == 1) {
                                    log.info(ip + "获得任务: " + task.getId() + " 开始执行!");

                                    ProcessorRunner runner = SpringContextUtil.getBean(ProcessorRunner.class);
                                    runner.setSemp(semp);
                                    runner.setTask(task);

                                    Future<?> future = executor.submit(runner);
                                    FutureHolder.holdFuture(task.getId(), future);
                                }

                            } else {
                                log.info(ip + " 检测锁定任务失败: " + task.getId() + " 放弃任务!") ;
                                semp.release();
                            }
                        } else if (count == 0) {
                            //锁定任务失败，放弃任务，重新选取下一个
                            log.info(ip + " lock task " + task.getId() + " 失败,放弃任务!");
                            semp.release();
                        }

                        ids = new LinkedList<Long>();

                    } else {
                        log.info("没有task，休眠: " + Constants.runNewJobInterval + " 毫秒");
                        Thread.sleep(Constants.runNewJobInterval);
                        semp.release();
                    }
                } else {
                    log.info("分发器没有开启，休眠{}毫秒", Constants.runNewJobInterval);
                    Thread.sleep(Constants.runNewJobInterval);
                }
            }

        } catch (InterruptedException e) {
            log.info("RunNewJob thread sleep exception!" + e.getMessage(), e);
            semp.release();
            try{
                Thread.sleep(Constants.runNewJobInterval);
            } catch (InterruptedException ine) {

            }
        }
    }

    private void idle() {
        int sum = 0;
        for(int i=0; i<1000000;i++) {
            sum = sum + i;
        }
    }
}